package com.thinkingcoder.tcmq.server;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import com.thinkingcoder.tcmq.core.model.Message;
import com.thinkingcoder.tcmq.core.model.Stat;
import com.thinkingcoder.tcmq.core.model.Subscription;
import com.thinkingcoder.tcmq.store.Entry;
import com.thinkingcoder.tcmq.store.Indexer;
import com.thinkingcoder.tcmq.store.Store;

import lombok.Getter;

/**
 * queue.
 */
public class MessageQueue {

    private static final String TEST_TOPIC = "com.thinkingcoder.tcmq.test";
    private static final String TEST_TC_DFS_TOPIC = "com.thinkingcoder.tcdfs.test";

    private String topic;

    // private Message<?>[] queue = new Message[10 * 1024];
    // private int writeOffset = 0;
    @Getter
    private Store store = null;

    private Map<String, Subscription> subscriptions = new HashMap<>();

    public static final Map<String, MessageQueue> queues = new ConcurrentHashMap<>(128);

    static {
        queues.put(TEST_TOPIC, new MessageQueue(TEST_TOPIC));
        queues.put(TEST_TC_DFS_TOPIC, new MessageQueue(TEST_TC_DFS_TOPIC));
        // queues.put("demo", new MessageQueue("demo"));
    }

    public MessageQueue(String topic) {
        this.topic = topic;
        store = new Store(topic);
        store.init();
    }

    public int doSend(Message<String> message) {
        int writeOffset = store.pos();
        message.getHeaders().put("X-offset", String.valueOf(writeOffset));
        store.write(message);
        return writeOffset;
    }

    public Message<?> doReceive(int readingOffset) {
        return store.read(readingOffset);
    }

    public void addSubscription(Subscription subscription) {
        String consumerId = subscription.getConsumerId();
        subscriptions.putIfAbsent(consumerId, subscription);
    }

    public void removeSubscription(Subscription subscription) {
        String consumerId = subscription.getConsumerId();
        subscriptions.remove(consumerId);
    }

    public static void subscribe(Subscription subscription) {
        MessageQueue messageQueue = queues.get(subscription.getTopic());
        System.out.println(" ===>>> subscribe: " + subscription);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }
        messageQueue.addSubscription(subscription);
    }

    public static void autoCreateTopic(String topic, boolean allowAutoCreate) {
        if (queues.containsKey(topic)) {
            return;
        }
        if (!allowAutoCreate) {
            throw new RuntimeException("disable auto create topic.");
        }
        queues.put(topic, new MessageQueue(topic));
    }

    public static void unSubscribe(Subscription subscription) {
        MessageQueue messageQueue = queues.get(subscription.getTopic());
        System.out.println(" ===>>> unSubscribe: " + subscription);
        if (messageQueue == null) {
            return;
        }
        messageQueue.removeSubscription(subscription);
    }

    public static int send(String topic, Message<String> message) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }
        System.out.println(" ===>>> send: topic/message" + topic + "/" + message);
        return messageQueue.doSend(message);
    }

    public static Message<?> receive(String topic, String consumerId, int readingIndex) {
        MessageQueue messageQueue = queues.get(topic);
        System.out
            .println(" ===>>> receive: topic/consumerId/readingIndex" + topic + "/" + consumerId + "/" + readingIndex);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("subscriptions not found for top/consumerId = " + topic + "/" + consumerId);
        }
        return messageQueue.doReceive(readingIndex);
    }

    // 使用此方法，需要手工调用ack，更新订阅关系里面的offset
    public static Message<?> receive(String topic, String consumerId) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("subscriptions not found for top/consumerId = " + topic + "/" + consumerId);
        }
        int readOffset = messageQueue.subscriptions.get(consumerId).getReadOffset();
        int readingOffset = 0;
        if (readOffset > -1) {
            Entry entry = Indexer.getEntry(topic, readOffset);
            readingOffset = readOffset + entry.getLength();
        }
        Message<?> message = messageQueue.doReceive(readingOffset);
        System.out
            .println(" ===>>> receive: topic/cid/readingOffset = " + topic + "/" + consumerId + "/" + readingOffset);
        System.out.println(" ===>>> receive: message = " + message);
        return message;
    }

    public static List<Message<?>> batch(String topic, String consumerId, int size) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("subscriptions not found for topic/consumerId = " + topic + "/" + consumerId);
        }
        int readOffset = messageQueue.subscriptions.get(consumerId).getReadOffset();
        int readingOffset = 0;
        if (readOffset > -1) {
            Entry entry = Indexer.getEntry(topic, readOffset);
            readingOffset = readOffset + entry.getLength();
        }
        List<Message<?>> result = new ArrayList<>();
        Message<?> receiveMsg = messageQueue.doReceive(readingOffset);
        while (receiveMsg != null) {
            result.add(receiveMsg);
            if (result.size() >= size) {
                break;
            }
            receiveMsg = messageQueue.doReceive(++readingOffset);
        }
        System.out.println(" ===>> batch: topic/cid/size = " + topic + "/" + consumerId + "/" + result.size());
        System.out.println(" ===>> last message: " + receiveMsg);
        return result;
    }

    public static int ack(String topic, String consumerId, int readIndex) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found.");
        }

        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("subscriptions not found for top/consumerId = " + topic + "/" + consumerId);
        }

        Subscription subscription = messageQueue.subscriptions.get(consumerId);
        if (readIndex > subscription.getReadOffset() && readIndex <= Store.LENGTH) {
            System.out.println(" ===>>> ack: topic/cid/offset = " + topic + "/" + consumerId + "/" + readIndex);
            subscription.setReadOffset(readIndex);
            return readIndex;
        }
        return -1;

    }

    public static Stat stat(String topic, String consumerId) {
        MessageQueue messageQueue = queues.get(topic);
        Subscription subscription = messageQueue.subscriptions.get(consumerId);
        return new Stat(subscription, messageQueue.getStore().total(), messageQueue.getStore().pos());
    }

}
